package com.dsk.connector.source.mysql.handler;

import cn.hutool.core.date.DatePattern;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONUtil;
import cn.hutool.log.StaticLog;
import com.dsk.connector.commons.conf.TableRegMapping;
import com.dsk.connector.commons.trans.FieldSchema;
import com.dsk.connector.commons.trans.RecordDdl;
import com.dsk.connector.commons.trans.RecordDml;
import com.dsk.connector.commons.trans.TransMsg;
import com.dsk.connector.commons.utils.DDLUtils;
import com.dsk.connector.commons.utils.SerializeUtils;
import com.dsk.connector.sink.common.listener.DatatransMqUtils;
import com.dsk.connector.source.common.DebeziumHandler;
import com.dsk.connector.source.utils.ChangeEventUtils;
import com.dsk.connector.source.utils.SourceRecordUtils;
import io.debezium.engine.RecordChangeEvent;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.*;

@Component("mysqlRecordHandler")
public class MysqlRecordHandler extends DebeziumHandler<RecordChangeEvent<SourceRecord>> {
    Logger logger= LoggerFactory.getLogger(MysqlRecordHandler.class);
    Logger sourcePoint= LoggerFactory.getLogger("sourcePoint");
    private ThreadLocal<TransMsg> msgThreadLocal=new ThreadLocal<TransMsg>();

    private Integer mqBatchSize;

    public int getBatchSize() {
        if(mqBatchSize!=null && mqBatchSize>0){
            return mqBatchSize;
        }
        try{
            String batchSizeStr=environment.getProperty("config.source.batchSize","20");
            this.mqBatchSize=Integer.valueOf(batchSizeStr);
            return this.mqBatchSize;
        }catch (Throwable ex){
            ex.printStackTrace();
            StaticLog.error(ex);
            this.mqBatchSize=20;
            return this.mqBatchSize;
        }

    }


    public void sendBatchMessage(){
        TransMsg transMsg=this.msgThreadLocal.get();
        if(transMsg!=null){
            this.msgThreadLocal.set(null);
            if(StringUtils.isBlank(transMsg.getTableName())){
                transMsg.setTableName("default");
            }
            if(transMsg.getDatas()!=null && transMsg.getDatas().size()>0){
                for(RecordDml dml:transMsg.getDatas()){
                    if(StringUtils.isBlank(dml.getTableName())){
                        dml.setTableName("default");
                    }
                }
            }
            String mqMsg= SerializeUtils.hessianSerialize(transMsg);
            String topic= DatatransMqUtils.getInstance().getQueueName(transMsg.getTableName(),transMsg.getOperation());

            this.sendMessage(topic,mqMsg);
        }
    }

    @Override
    public void handleRecord(RecordChangeEvent<SourceRecord> event, int processedCount, int batchSize) {
        SourceRecord record=event.record();
//        System.out.println("Key = '" + record.key() );
//        System.out.println("value = '" + record.value() + "'");
        logger.debug("Key = '" + record.key() );
        logger.debug("value = '" + record.value() + "'");
        if(record.value()==null){
            this.sendBatchMessage();
            //删除操作时发送两个事件，第二个事件，value为空; 因为第一条已处理，因此忽略第二次通知
            return ;
        }
        Struct valueStruct=(Struct)record.value();

        sourcePoint.info("receive size:"+batchSize+",sourcePoint:"+ JSONUtil.toJsonStr(this.getSourcePoint(valueStruct),new JSONConfig().setDateFormat(DatePattern.NORM_DATETIME_PATTERN)));
        if(valueStruct.schema().field("ddl")!=null){
            this.handleDdlEvent(record,processedCount,batchSize);
        }else{
            String op= valueStruct.getString("op");
            if("t".equals(op)) {
                this.handleTruncate(record,processedCount,batchSize);
            }else{
                this.handleDmlEvent(record,processedCount,batchSize);
            }
        }
        if(processedCount%this.getBatchSize()==0){
            this.sendBatchMessage();
        }
        if(processedCount>=batchSize){
            this.sendBatchMessage();
        }

    }

    private void handleDmlEvent(SourceRecord record, int processedCount, int batchSize) {
        //如果开启了表名正则匹配，则根据正则匹配表名，如果匹配不到则忽略
        if(this.isRegTableName()){
            String tableName= ChangeEventUtils.retriveTableNamefromKey(record.topic());
            if(!this.getTableMappings().containsKey(tableName)){
                //如果配置了正则表格式，则根据正则匹配表名
                if(this.getTableRegMappings().size()>0){
                    for(TableRegMapping regMapping:this.getTableRegMappings().values()){
                        TableRegMapping mapping=regMapping.getMatchMapping(tableName);
                        if(mapping!=null){
                            this.getTableMappings().put(tableName,mapping);
                            break;
                        }
                    }
                }
                //无法匹配到该表，忽略
                if(!this.getTableMappings().containsKey(tableName)){
                    if(processedCount==batchSize){
                        this.sendBatchMessage();
                    }
                    return ;
                }
            }

        }

        //匹配表名成功，则进行后续操作
        RecordDml dmlRecord=null;
        Struct valueStruct=(Struct)record.value();
        String op= valueStruct.getString("op");
        if("c".equals(op)||"r".equals(op)){
            dmlRecord=this.handleInsert(record);
        }
        if("u".equals(op)){
            dmlRecord=this.handleUpdate(record);
        }
        if("d".equals(op)){
            dmlRecord=this.handleDelete(record);
        }

        if(dmlRecord==null){
            logger.error("op is not support:key={},value={}",record.key(),record.value());
            this.sendBatchMessage();
            return;
        }
        if(this.msgThreadLocal.get()==null){
            TransMsg transMsg=this.createNewTrans(dmlRecord,this.getSourcePoint(valueStruct));
            this.msgThreadLocal.set(transMsg);
        }else{
            TransMsg transMsg=this.msgThreadLocal.get();
            if(transMsg.canAdd(dmlRecord)){
                transMsg.addRecord(dmlRecord);
                this.msgThreadLocal.set(transMsg);
            }else{
                this.sendBatchMessage();
                transMsg=this.createNewTrans(dmlRecord,this.getSourcePoint(valueStruct));
                this.msgThreadLocal.set(transMsg);
            }
        }
        if(processedCount==batchSize){
            this.sendBatchMessage();
        }
    }
    private Map<String,Object> getSourcePoint(Struct valueStruct){
        Map<String,Object> sourcePoint=new HashMap<>();
        if(valueStruct.schema().field("source")!=null){
            Struct source=valueStruct.getStruct("source");
            sourcePoint=SourceRecordUtils.convertFieldValues(source);
            if(source.schema().field("ts_ms")!=null && sourcePoint.get("ts_ms") instanceof Long){
                sourcePoint.put("ts_ms",new Date((Long)sourcePoint.get("ts_ms")));
            }
        }
        return sourcePoint;
    }

    private RecordDml handleTruncate(SourceRecord record,int processedCount, int batchSize) {
        //把遗留的数据先发送出去
        this.sendBatchMessage();
        //
        Struct valueStruct=(Struct)record.value();
        Struct source=valueStruct.getStruct("source");
        String databaseName=source.getString("db");
        String table=source.getString("table");

        String ddl="truncate table "+databaseName+"."+table;
        if(valueStruct.schema().field("ddl")!=null){
            ddl=valueStruct.getString("ddl");
        }
        //组织接下来的发送数据
        TransMsg transMsg=new TransMsg();
        transMsg.setTableName(databaseName);
        transMsg.setOperation("ddl");
        transMsg.setSourcePoint(this.getSourcePoint(valueStruct));
        String type="truncate";
        String tableName=databaseName+"."+table;
        if(StringUtils.isBlank(databaseName) && StringUtils.isBlank(table)){
            tableName="default";
            transMsg.setTableName(tableName);
        }
        RecordDdl ddlRecord=new RecordDdl();
        ddlRecord.setOperation(type.toLowerCase());
        ddlRecord.setAlterSql(ddl);
        ddlRecord.setTableName(tableName);
        transMsg.addRecord(ddlRecord);

        this.msgThreadLocal.set(transMsg);
        //发送alter信息
        this.sendBatchMessage();

        return null;
    }

    private RecordDml handleDelete(SourceRecord record) {
        RecordDml delete=new RecordDml();
        delete.setOperation("delete");
        String tableName= ChangeEventUtils.retriveTableNamefromKey(record.topic());
        delete.setTableName(tableName);

        Struct keyStruct=(Struct)record.key();
        if(keyStruct==null){
            Struct valueStruct=(Struct)record.value();
            //针对无主键的表，直接发送全字段数据
            if(record.valueSchema().field("before")!=null && valueStruct.get("before")!=null){
                //针对修改操作
                List<FieldSchema> keyFieldsSchema=SourceRecordUtils.convertToFieldSchema(record.valueSchema().field("before").schema());
                delete.setKeySchemas(keyFieldsSchema);
                delete.setPayloadSchemas(keyFieldsSchema);
                Map<String,Object> keyValues=SourceRecordUtils.convertFieldValues((Struct)valueStruct.get("before"));
                delete.setKeyValue(keyValues);
                delete.setPayloadValues(keyValues);
            }else{
                return null;
            }
        }else{
            //设置key部分
            List<FieldSchema> keyFieldsSchema=SourceRecordUtils.convertToFieldSchema(record.keySchema());
            delete.setKeySchemas(keyFieldsSchema);
            Map<String,Object> keyValues=SourceRecordUtils.convertFieldValues(keyStruct);
            delete.setKeyValue(keyValues);
            delete.setPayloadValues(keyValues);
            delete.setPayloadSchemas(keyFieldsSchema);
        }


        return delete;
    }

    private RecordDml handleUpdate(SourceRecord record) {
        RecordDml update=this.handleInsert(record);
        update.setOperation("update");
        return update;
    }

    private void handleDdlEvent(SourceRecord record, int processedCount, int batchSize) {
        //把遗留的数据先发送出去
        this.sendBatchMessage();
        //
        Struct valueStruct=(Struct)record.value();
        String ddl= valueStruct.getString("ddl");
        logger.info("ddl={}",ddl);
        List<Struct> tableChanges=valueStruct.getArray("tableChanges");
        String databaseName=valueStruct.getString("databaseName");
        //组织接下来的发送数据
        TransMsg transMsg=new TransMsg();
        transMsg.setTableName(databaseName);
        transMsg.setOperation("ddl");
        transMsg.setSourcePoint(this.getSourcePoint(valueStruct));
        if(ddl.toLowerCase().trim().startsWith("truncate")){
            this.handleTruncate(record,processedCount,batchSize);
            return;
        }else{
            for(Struct changeTable:tableChanges){
                String type=changeTable.getString("type");
                String tableName=changeTable.getString("id");
                tableName=tableName.replaceAll("\"","");
                RecordDdl ddlRecord=new RecordDdl();
                ddlRecord.setOperation(type.toLowerCase());
                ddlRecord.setAlterSql(ddl);
                ddlRecord.setTableName(tableName);
                transMsg.addRecord(ddlRecord);
            }
        }
        if(transMsg.getDatas()==null || transMsg.getDatas().size()==0){
            System.out.println("ddl no payload:"+record);
            logger.error("ddl no payload:{}",record.toString());
            if(valueStruct.schema().field("source")!=null){
                Struct sourceStruct=(Struct)valueStruct.getStruct("source");
                String db=sourceStruct.getString("db");
                String table=sourceStruct.getString("table");
                String tableName=db;
                if(StringUtils.isNotBlank(table)){
                    tableName=tableName+"."+table;
                }
                if(StringUtils.isBlank(tableName)){
                    tableName="default";
                }
                ddl= DDLUtils.removeComment(ddl);
                String operation=ddl.substring(0,ddl.indexOf(" "));
                RecordDdl ddlRecord=new RecordDdl();
                ddlRecord.setOperation(operation.toLowerCase());
                ddlRecord.setAlterSql(ddl);
                ddlRecord.setTableName(tableName);
                transMsg.addRecord(ddlRecord);
            }
        }

        if(StringUtils.isBlank(transMsg.getTableName())){
            transMsg.setTableName("default");
        }
        this.msgThreadLocal.set(transMsg);
        //发送alter信息
        this.sendBatchMessage();

    }

    public TransMsg createNewTrans(RecordDml dmlRecord,Map<String,Object> sourcePoint){
        TransMsg transMsg=new TransMsg();
        transMsg.setDatas(Arrays.asList(dmlRecord));
        transMsg.setOperation(dmlRecord.getOperation());
        transMsg.setTableName(dmlRecord.getTableName());
        transMsg.setModifyFields(dmlRecord.getModifyFields());
        transMsg.setSourcePoint(sourcePoint);

        return transMsg;
    }

    private RecordDml handleInsert(SourceRecord record) {
        RecordDml insert=new RecordDml();
        insert.setOperation("insert");
        String tableName= ChangeEventUtils.retriveTableNamefromKey(record.topic());
        insert.setTableName(tableName);
        //设置value部分
        Struct valueStruct=(Struct)record.value();
        List<FieldSchema> payloadFieldsSchema=SourceRecordUtils.convertToFieldSchema(record.valueSchema().field("after").schema());
        insert.setPayloadSchemas(payloadFieldsSchema);
        Map<String,Object> payloadValues=SourceRecordUtils.convertFieldValues((Struct)valueStruct.get("after"));
        insert.setPayloadValues(payloadValues);
        //设置key部分
        Struct keyStruct=(Struct)record.key();
        if(keyStruct==null){
            //针对无主键的表处理
            if(record.valueSchema().field("before")!=null && valueStruct.get("before")!=null){
                //针对修改操作
                List<FieldSchema> beforeFieldsSchema=SourceRecordUtils.convertToFieldSchema(record.valueSchema().field("before").schema());
                insert.setKeySchemas(beforeFieldsSchema);
                Map<String,Object> beforeValues=SourceRecordUtils.convertFieldValues((Struct)valueStruct.get("before"));
                insert.setKeyValue(beforeValues);
            }else{
                //针对新增操作
                insert.setKeySchemas(insert.getPayloadSchemas());
                insert.setKeyValue(insert.getPayloadValues());
            }

        }else{
            List<FieldSchema> keyFieldsSchema=SourceRecordUtils.convertToFieldSchema(record.keySchema());
            insert.setKeySchemas(keyFieldsSchema);
            Map<String,Object> keyValues=SourceRecordUtils.convertFieldValues(keyStruct);
            insert.setKeyValue(keyValues);
        }

        return insert;
    }

    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }
}
